package avicit.bdp.dds.api.service;

import avicit.bdp.common.dto.StorageResourceConf;
import avicit.bdp.common.service.service.CalculateEngineConfigService;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.uploads.FileAdapterUtils;
import avicit.bdp.common.utils.uploads.IFileAdapter;
import avicit.bdp.core.http.HttpClientUtil;
import avicit.bdp.core.http.common.HttpConfig;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.job.WebSocketProcessStausServer;
import avicit.bdp.dds.api.job.WebSocketTaskLogServer;
import avicit.bdp.dds.api.job.WebSocketTaskStatusServer;
import avicit.bdp.dds.api.utils.TaskStateUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.utils.HdfsUtils;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.dao.entity.TaskLog;
import avicit.bdp.dds.dao.mapper.ProcessDefinitionMapper;
import avicit.bdp.dds.dao.mapper.ProcessInstanceMapper;
import avicit.bdp.dds.dao.mapper.ProjectMapper;
import avicit.bdp.dds.dao.mapper.TaskInstanceMapper;
import avicit.bdp.dds.remote.utils.Host;
import avicit.bdp.dds.service.log.LogClientService;
import avicit.platform6.core.rest.msg.ResponseMsg;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import javax.websocket.Session;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author admin
 */
@Service
public class LoggerService {

	private static final Logger logger = LoggerFactory.getLogger(LoggerService.class);
	@Autowired
	private TaskInstanceMapper taskInstanceMapper;
	@Autowired
	private ProcessInstanceMapper processInstanceMapper;
	@Autowired
	private ProcessDefinitionMapper processDefinitionMapper;
	@Autowired
	private ProjectMapper ProjectMapper;
	@Autowired
	private CalculateEngineConfigService calculateEngineConfigService;
	private LogClientService logClient;
	@Autowired
	private TaskLogService taskLogService;
	@Autowired
	private WebSocketTaskLogServer webSocketTaskLogServer;
	@Autowired
	private WebSocketTaskStatusServer webSocketTaskStatusServer;
	@Autowired
	private WebSocketProcessStausServer webSocketProcessStausServer;


	public LoggerService() {
		logClient = new LogClientService();
	}

	@PreDestroy
	public void close() {
		logClient.close();
	}

	/**
	 * view log
	 *
	 * @param taskInstId  task instance id
	 * @param skipLineNum skip line number
	 * @param limit       limit
	 * @return log string data
	 */
	public ResponseMsg<String> queryLog(String taskInstId, int skipLineNum, int limit) {
		ResponseMsg<String> responseMsg = new ResponseMsg<>();
		if (StringUtils.isEmpty(taskInstId)) {
			throw new IllegalArgumentException("请检查查询参数。");
		}

		TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstId);
		if (taskInstance == null || StringUtils.isBlank(taskInstance.getHost())) {
			logger.error("the task instance is null.");
			responseMsg.setRetCode(String.valueOf(500));
			responseMsg.setErrorDesc("the task instance is null.");
			return responseMsg;
		}

		String host = getHost(taskInstance.getHost());

		logger.info("log host : {} , logPath : {} , logServer port : {}", host, taskInstance.getLogPath(), Constants.RPC_PORT);

		String log = logClient.rollViewLog(host, Constants.RPC_PORT, taskInstance.getLogPath(), skipLineNum, limit);
		responseMsg.setResponseBody(log);

		return responseMsg;
	}

	/**
	 * view log
	 *
	 * @param taskInstId task instance id
	 * @return log string data
	 */
	public ResponseMsg<String> queryAllLog(String taskInstId) {
		ResponseMsg<String> responseMsg = new ResponseMsg<>();
		if (StringUtils.isEmpty(taskInstId)) {
			throw new IllegalArgumentException("请检查查询参数。");
		}

		TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstId);
		if (taskInstance == null || StringUtils.isBlank(taskInstance.getHost())) {
			logger.error("the task instance is null.");
			responseMsg.setResponseBody("任务正在提交中");
			return responseMsg;
		}

		String host = getHost(taskInstance.getHost());

		logger.info("log host : {} , logPath : {} , logServer port : {}", host, taskInstance.getLogPath(), Constants.RPC_PORT);

		String log = logClient.viewLog(host, Constants.RPC_PORT, taskInstance.getLogPath());
		responseMsg.setResponseBody(log);

		return responseMsg;
	}

	public ResponseMsg<String> queryAllLogLimit(String taskInstId, int skipLineNum, int limit) {
		ResponseMsg<String> responseMsg = new ResponseMsg<>();
		if (StringUtils.isEmpty(taskInstId)) {
			throw new IllegalArgumentException("请检查查询参数。");
		}

		TaskLog taskLog = taskLogService.selectByPrimaryKey(taskInstId);
		if (taskLog == null) {
			responseMsg.setResponseBody("没有日志信息");
		} else {
			Stream<String> info = Arrays.stream(taskLog.getContents().split("\n"));
			List<String> str = info.skip(skipLineNum).limit(limit).collect(Collectors.toList());
			responseMsg.setResponseBody(String.join("\n", str));
		}

		return responseMsg;
	}


	/**
	 * get log size
	 *
	 * @param taskInstId task instance id
	 * @return log byte array
	 */
	public byte[] getLogBytes(String taskInstId) {
		TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstId);
		if (taskInstance == null || StringUtils.isBlank(taskInstance.getHost())) {
			throw new RuntimeException("task instance is null or host is null");
		}
		String host = getHost(taskInstance.getHost());

		return logClient.getLogBytes(host, Constants.RPC_PORT, taskInstance.getLogPath());
	}


	/**
	 * get host
	 *
	 * @param address address
	 * @return old version return true ,otherwise return false
	 */
	private String getHost(String address) {
		if (Host.isOldVersion(address)) {
			return address;
		}
		return Host.of(address).getIp();
	}


	public String getLogByTaskInstanceId(String taskId, String type) {
		ProcessInstance instance = processInstanceMapper.getProcessInstanceByTaskId(taskId);

		String exparams = instance.getExtraParams();
		String appid = "";
		if (StringUtils.isNotBlank(exparams)) {
			Map<String, Object> paramMap = JSONUtils.toObjectMap(exparams);
			appid = MapUtils.getString(paramMap, "appid");
		}

		String content = "";

		if ("localLog".equals(type)) {
			TaskLog taskLog = taskLogService.selectByPrimaryKey(instance.getId());
			if (taskLog != null) {
				content = taskLog.getContents();
			}
			return content;
		}

		if (StringUtils.isBlank(appid)) {
			return "";
		}

		String engineId = ProjectMapper.getResourceEngineIdByTaskId(taskId);
		String params = "";
		if (StringUtils.isNotBlank(engineId)) {
			params = calculateEngineConfigService.getParamsById(engineId);
		} else {
			params = calculateEngineConfigService.getDefaultSparkParams();
		}
		String yarnUrl = "";
		if (StringUtils.isNotBlank(params)) {
			Map<String, Object> map = JSONUtils.toObjectMap(params);
			yarnUrl = MapUtils.getString(map, "yarnAddress");
		}
		yarnUrl = yarnUrl + "/ws/v1/cluster/apps/" + appid;

		Map<String, Object> result;
		HttpConfig config = HttpConfig.custom().timeout(1000).url(yarnUrl);

		try {
			try {
				String responseContent = HttpClientUtil.get(config);
				result = JSONUtils.toObjectMap(responseContent);
			} catch (Exception e) {
				throw new RuntimeException("获取yarn日志异常" + e.getMessage());
			}

			Map<String, Object> app = (Map<String, Object>) MapUtils.getMap(result, "app");
			String amContainerLogs = MapUtils.getString(app, "amContainerLogs");
			logger.info(amContainerLogs);

			if (StringUtils.isNotBlank(amContainerLogs)) {
				//rtnMap.put("stderrLog", amContainerLogs + "/stderr/?start=0");
				//spark 应用日志
				if ("app".equals(type)) {
					String outUrl = amContainerLogs + "/stdout/?start=-1048576";
					content = content + getLogDetail(outUrl);
				} else if ("run".equals(type)) {
					//spark 运行日志
					String errUrl = amContainerLogs + "/stderr/?start=-1048576";
					content = content + getLogDetail(errUrl);
				} else {
					String errUrl = amContainerLogs + "/stderr/?start=-1048576";
					String outUrl = amContainerLogs + "/stdout/?start=-1048576";
					content = content + getLogDetail(outUrl);
					content = content + "\n\n";
					content = content + getLogDetail(errUrl);
				}
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}

		return content;
	}

	private String getLogDetail(String url) {
		HttpConfig config = HttpConfig.custom().timeout(1000).url(url);
		try {
			String content = HttpClientUtil.get(config);
			Document doc = Jsoup.parse(content);
			Elements e = doc.getElementsByClass("content");
			return e.html();
		} catch (Exception e) {
			throw new RuntimeException("获取yarn日志异常" + e.getMessage());
		}
	}

    public List<Map<String, Object>> queryFlowData(String taskInstanceId) {
        TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstanceId);
		ProcessDefinition processDefinition = processDefinitionMapper.selectByPrimaryKey(taskInstance.getProcessDefinitionId());
        //String content = FlowUtils.getDebugDataAll(taskInstance.getAppLink(), taskInstance.getName());
        String debugPath = "/user/piflow/debug/" + taskInstance.getAppLink() + "/" + taskInstance.getName();
        //List<String> list = HdfsUtils.getDirs(debugPath, taskInstance.getProcessDefinitionId());
		String projectId = processDefinition.getProjectId();
		StorageResourceConf storageResourceConf = FileAdapterUtils.getStorageResourceByProjectId(projectId);
		IFileAdapter fileAdapter = FileAdapterUtils.getFileService(storageResourceConf);
		List<String> list = fileAdapter.getDirs(debugPath);
        List<String> ports = new ArrayList<>();
        for (String dir : list) {
            if (!dir.endsWith("_schema")) {
                ports.add(dir.substring(dir.lastIndexOf("/") + 1));
            }
        }
        List<Map<String, Object>> resultList = new ArrayList<>();
        for (String port : ports) {
            String path = debugPath + "/" + port;
            //String schema = HdfsUtils.getLine(path + "_schema", taskInstance.getProcessDefinitionId());
			String schema = fileAdapter.readText(path + "_schema");
            schema = schema.replaceAll("[\\n\\r]", "");
            //List<Map<String, Object>> data = HdfsUtils.getJsonMapList(path, taskInstance.getProcessDefinitionId());
			List<Map<String, Object>> data = fileAdapter.readJsonFile(path);
            Map<String, Object> result = new HashMap<>();
            result.put("port", port);
            result.put("schema", schema);
            result.put("debugDataPath", path);
            result.put("data", data);
            resultList.add(result);
        }

		return resultList;
	}

	public Map<String, Object> queryFlowAnalysisData(String taskInstanceId) {
		TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstanceId);
		if (taskInstance == null) {
			return new HashMap<>();
		}
		ProcessDefinition processDefinition = processDefinitionMapper.selectByPrimaryKey(taskInstance.getProcessDefinitionId());

		String path = "/user/piflow/visual/" + taskInstance.getAppLink() + "/" + taskInstance.getName() + "/data";
		String projectId = processDefinition.getProjectId();
		StorageResourceConf storageResourceConf = FileAdapterUtils.getStorageResourceByProjectId(projectId);
		IFileAdapter fileAdapter = FileAdapterUtils.getFileService(storageResourceConf);
        String schema = "";
        List<Map<String, Object>> list = new ArrayList<>();
        Map<String, Object> schemaMap = new HashMap<>();
        try {
            //list = HdfsUtils.getParquetMapList(path, taskInstance.getProcessDefinitionId());
			list = fileAdapter.getParquetMapList(path);
            String schemaPath = "/user/piflow/visual/" + taskInstance.getAppLink() + "/" + taskInstance.getName() + "/schema";
            schema = fileAdapter.readText(schemaPath);

			schemaMap = JSONUtils.toObjectMap(schema);
			//schemaMap.put("chartType","MultiLineChart");
			String chartType = MapUtils.getString(schemaMap, "chartType", "");

			//统一处理诊断结果
			//handleDiagnosisResult(list, schemaMap);

			//处理各个图形的显示数据
			if ("LineChart".equalsIgnoreCase(chartType)) {
				handleLineTypeChart(list, schemaMap);
			} else if ("MultiLineChart".equalsIgnoreCase(chartType)) {
				handleMultiLineTypeChart(list, schemaMap);
			} else if ("ScatterChart".equalsIgnoreCase(chartType)) {
				handleScatterTypeChart(list, schemaMap);
			} else if ("FittingChart".equalsIgnoreCase(chartType)) {
				handleFittingeChart(list, schemaMap);
			} else if ("BarChart".equalsIgnoreCase(chartType)) {
				handleBarChart(list, schemaMap);
			} else if ("TableChart".equalsIgnoreCase(chartType)) {
				handleTableChart(list, schemaMap);
			} else if ("PythonChart".equalsIgnoreCase(chartType)) {
				//不用处理，只返回map里的图片路径
			}
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(e.getMessage(), e);

		}

		return schemaMap;
	}

	private void handleDiagnosisResult(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		//处理诊断结果， 根据列名，获取第一条记录返回诊断结果
		String diagnosisParam = MapUtils.getString(schemaMap, "diagnosisParam", "");
		if (!list.isEmpty() && StringUtils.isNotBlank(diagnosisParam)) {
			String result = MapUtils.getString(list.get(0), diagnosisParam, "");
			schemaMap.put("diagnosisResult", result);
		}
	}

	private void handleTableChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		schemaMap.put("data", list);
	}

	private void handleBarChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		// schemaMap
		// {"chartType":"BarChart","title":"","legend":true,"diagnosisParam","",chartList":[{
		//  x":"","y":[{"name":"","color":"","cnname":""}]}]}
		// 柱状图
		List<Map<String, Object>> chartList = (List<Map<String, Object>>) schemaMap.get("chartList");
		List<Map<String, Object>> dataList = new ArrayList<>();
		for (Map<String, Object> chart : chartList) {
			Map<String, Object> data = new HashMap<>();
			String title = MapUtils.getString(chart, "title", "");
			data.put("title", title);
			data.put("key", MapUtils.getString(chart, "key"));
			Map<String, Object> xdata = new HashMap<>();
			List<Object> xlist = new ArrayList<>();
			String xLabel = MapUtils.getString(chart, "x", "");
			String yLabel = MapUtils.getString(chart, "y", "");

			List<Map<String, Object>> yList = (List<Map<String, Object>>) chart.get("y");
			List<Map<String, Object>> ydataList = new ArrayList<>();
			List<String> xnameList = new ArrayList<>();
			//x轴 名称数据
			for (Map<String, Object> yMap : yList) {
				String key = MapUtils.getString(yMap, "cnname");
				if (StringUtils.isBlank(key)) {
					key = MapUtils.getString(yMap, "name");
				}
				xnameList.add(key);
			}

			int count = 1;

			for (Map<String, Object> map : list) {
				Object x = MapUtils.getObject(map, xLabel);
				Map<String, Object> cdata = new HashMap<>();
				//分类数据
				cdata.put("name", x);
				List<Object> clist = new ArrayList<>();
				//处理y轴数据
				for (Map<String, Object> yMap : yList) {
					String key = MapUtils.getString(yMap, "name");
					Object y = MapUtils.getObject(map, key);
					clist.add(y);
				}
				cdata.put("data", clist);
				xlist.add(cdata);
				count++;
			}
			xdata.put("data", xlist);
			xdata.put("xnameList", xnameList);

			dataList.add(xdata);
		}
		schemaMap.put("data", dataList);
	}

	private void handleFittingeChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		// schemaMap
		// {"chartType":"FittingChart","title":"","diagnosisParam","","legend":true,"crossLine":true,"smooth":true,chartList":[{
		// "xname":"123",x":"","y":"","ycolor":"","linexName":"","linex":"","liney":"","lineyColor":""}]}
		// 拟合图
		List<Map<String, Object>> chartList = (List<Map<String, Object>>) schemaMap.get("chartList");
		List<Map<String, Object>> dataList = new ArrayList<>();

		for (Map<String, Object> chart : chartList) {
			Map<String, Object> data = new HashMap<>();
			String title = MapUtils.getString(chart, "title", "");
			data.put("title", title);
			Map<String, Object> xdata = new HashMap<>();
			List<Object> xlist = new ArrayList<>();
			List<Object> alist = new ArrayList<>();
			xdata.put("xname", MapUtils.getString(chart, "xname", ""));
			xdata.put("linexName", MapUtils.getString(chart, "linexName", ""));
			String xLabel = MapUtils.getString(chart, "x", "");
			String yLabel = MapUtils.getString(chart, "y", "");
			String linexLabel = MapUtils.getString(chart, "linex", "");
			String lineyLabel = MapUtils.getString(chart, "liney", "");
			List<Object> linexList = new ArrayList<>();
			List<Object> lineyList = new ArrayList<>();
			double max_x = 0;
			double min_x = 0;
			double max_y = 0;
			double min_y = 0;
			int count = 1;
			for (Map<String, Object> map : list) {
				//散点图数据
				xlist = new ArrayList<>();
				Object x = MapUtils.getObject(map, xLabel);
				Object y = MapUtils.getObject(map, yLabel);
				xlist.add(x);
				xlist.add(y);
				alist.add(xlist);
				//计算 x，y最大、最小值
				double current_x = Double.parseDouble(x.toString());
				double current_y = Double.parseDouble(y.toString());
				if (current_x > max_x) {
					max_x = current_x;
				}
				if (current_x < min_x) {
					min_x = current_x;
				}
				if (current_y > max_y) {
					max_y = current_y;
				}
				if (current_y < min_y) {
					min_y = current_y;
				}
				//曲线图
				Object linex = MapUtils.getObject(map, linexLabel);
				Object liney = MapUtils.getObject(map, lineyLabel);
				linexList.add(linex);
				lineyList.add(liney);
				count++;
			}
			int abs_x = (int) Math.floor(Math.abs(Math.max(Math.abs(max_x), Math.abs(min_x))));
			int abs_y = (int) Math.floor(Math.abs(Math.max(Math.abs(max_y), Math.abs(min_y))));
			Map<String, Object> scatterData = new HashMap<>();
			scatterData.put("max_x", max_x);
			scatterData.put("min_x", min_x);
			scatterData.put("max_y", max_y);
			scatterData.put("min_y", min_y);
			scatterData.put("abs_x", abs_x);
			scatterData.put("abs_y", abs_y);
			xdata.put("scatterData", scatterData);
			xdata.put("data", alist);
			xdata.put("dataLinex", linexList);
			xdata.put("dataLiney", lineyList);

			dataList.add(xdata);
		}

		schemaMap.put("data", dataList);
	}

	private void handleScatterTypeChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		// schemaMap
		// {"chartType":"ScatterChart","title":"","diagnosisParam","","legend":true,"crossLine":true,chartList":[{
		// "xname":"123",x":"","y":"","ycolor":""}]}
		//散点图
		List<Map<String, Object>> chartList = (List<Map<String, Object>>) schemaMap.get("chartList");
		List<Map<String, Object>> dataList = new ArrayList<>();
		double max_x = 0;
		double min_x = 0;
		double max_y = 0;
		double min_y = 0;
		for (Map<String, Object> chart : chartList) {
			Map<String, Object> data = new HashMap<>();
			String title = MapUtils.getString(chart, "title", "");
			data.put("title", title);
			data.put("key", MapUtils.getString(chart, "key"));
			Map<String, Object> xdata = new HashMap<>();
			List<Object> xlist = new ArrayList<>();
			List<Object> alist = new ArrayList<>();
			xdata.put("xname", MapUtils.getString(chart, "xname", ""));
			String xLabel = MapUtils.getString(chart, "x", "");
			String yLabel = MapUtils.getString(chart, "y", "");

			int count = 1;
			for (Map<String, Object> map : list) {
				xlist = new ArrayList<>();
				Object x = MapUtils.getObject(map, xLabel);
				Object y = MapUtils.getObject(map, yLabel);
				xlist.add(x);
				xlist.add(y);
				alist.add(xlist);
				count++;
				//计算 x，y最大、最小值
				double current_x = Double.parseDouble(x.toString());
				double current_y = Double.parseDouble(y.toString());
				if (current_x > max_x) {
					max_x = current_x;
				}
				if (current_x < min_x) {
					min_x = current_x;
				}
				if (current_y > max_y) {
					max_y = current_y;
				}
				if (current_y < min_y) {
					min_y = current_y;
				}
			}
			xdata.put("data", alist);

			dataList.add(xdata);
		}
		int abs_x = (int) Math.floor(Math.abs(Math.max(Math.abs(max_x), Math.abs(min_x))));
		int abs_y = (int) Math.floor(Math.abs(Math.max(Math.abs(max_y), Math.abs(min_y))));
		Map<String, Object> scatterData = new HashMap<>();
		scatterData.put("max_x", max_x);
		scatterData.put("min_x", min_x);
		scatterData.put("max_y", max_y);
		scatterData.put("min_y", min_y);
		scatterData.put("abs_x", abs_x);
		scatterData.put("abs_y", abs_y);

		schemaMap.put("scatterData", scatterData);
		schemaMap.put("data", dataList);
	}

	private void handleMultiLineTypeChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		// schemaMap
		// {"chartType":"MultiLineChart","title":"","diagnosisParam","","legend":true,"crossLine":true,"smooth":true,chartList":[{
		// "autox":true,"xname":"123",x":"","y":[{"name":"","color":"","cnname":""}]}]}
		// 曲线图
		List<Map<String, Object>> chartList = (List<Map<String, Object>>) schemaMap.get("chartList");
		List<Map<String, Object>> dataList = new ArrayList<>();
		for (Map<String, Object> chart : chartList) {
			Map<String, Object> data = new HashMap<>();
			String title = MapUtils.getString(chart, "title", "");
			data.put("title", title);
			data.put("key", MapUtils.getString(chart, "key"));
			Map<String, Object> xdata = new HashMap<>();
			List<Object> xlist = new ArrayList<>();
			xdata.put("xname", MapUtils.getString(chart, "xname", ""));
			String xLabel = MapUtils.getString(chart, "x", "");
			boolean autox = MapUtils.getBooleanValue(chart, "autox", false);

			List<Map<String, Object>> yList = (List<Map<String, Object>>) chart.get("y");
			List<Map<String, Object>> ydataList = new ArrayList<>();
			//初始化y轴特征
			for (Map<String, Object> yMap : yList) {
				Map<String, Object> ydata = new HashMap<>();
				ydata.putAll(yMap);
				ydata.put("data", new ArrayList<>());
				ydataList.add(ydata);
			}
			int count = 1;
			for (Map<String, Object> map : list) {
				//处理x轴分类数据，如果自动生成，则以数据行数为x轴
				if (autox) {
					Object x = MapUtils.getObject(map, xLabel);
					xlist.add(x);
				} else {
					xlist.add(count);
				}

				//处理y轴特征
				for (Map<String, Object> yMap : yList) {
					String key = MapUtils.getString(yMap, "name");
					Object y = MapUtils.getObject(map, key);
					handleYData(ydataList, key, y);
				}

				count++;

			}
			xdata.put("data", xlist);
			data.put("x", xdata);
			data.put("y", ydataList);
			dataList.add(data);
		}
		schemaMap.put("data", dataList);

	}

	/**
	 * 简单折线图数据
	 */
	private Map<String, Object> handleLineTypeChart(List<Map<String, Object>> list, Map<String, Object> schemaMap) {
		String labels = MapUtils.getString(schemaMap, "labels", "");
		String features = MapUtils.getString(schemaMap, "features", "");
		//TODO 如果以后有其他的图形类型，另做处理数据结构
		String feature[] = features.split(",");
		List<Object> xlist = new ArrayList<>();
		Map<String, Object> data = new HashMap<>();
		List<Map<String, Object>> ydataList = new ArrayList<>();
		//初始化y轴特征
		for (String key : feature) {
			Map<String, Object> ydata = new HashMap<>();
			ydata.put("name", key);
			ydata.put("data", new ArrayList<>());
			ydataList.add(ydata);
		}

		for (Map<String, Object> map : list) {
			//处理x轴分类
			Object x = MapUtils.getObject(map, labels);
			xlist.add(x);
			//处理y轴特征
			for (String key : feature) {
				Object y = MapUtils.getObject(map, key);
				handleYData(ydataList, key, y);
			}

		}

		Map<String, Object> xdata = new HashMap<>();
		xdata.put("name", labels);
		xdata.put("data", xlist);
		data.put("x", xdata);
		data.put("y", ydataList);
		schemaMap.put("data", data);
		return data;
	}

	private void handleYData(List<Map<String, Object>> ydataList, String key, Object y) {
		for (Map<String, Object> map : ydataList) {
			String name = MapUtils.getString(map, "name");
			if (name.equalsIgnoreCase(key)) {
				List<Object> list = (List<Object>) MapUtils.getObject(map, "data");
				list.add(y);
				break;
			}
		}
	}

	public void insertTaskLog(TaskLog taskLog) {
		String taskName = taskLog.getTaskName();
		if (StringUtils.isBlank(taskName)) {
			taskName = "";
		}
		String contents = DateUtils.getCurrentTime() + " " + taskName + " " + taskLog.getContents() + "\n";
		taskLogService.updateTaskLog(taskLog.getId(), contents);
		String taskType = taskLog.getTaskType();
		String processDefinitionId = "";
		try {
			if (StringUtils.isNotBlank(taskType) && "task".equals(taskType)) {
				TaskInstance instance = taskInstanceMapper.selectByPrimaryKey(taskLog.getId());
				if (instance != null) {
					processDefinitionId = instance.getProcessDefinitionId();
				}
			} else {
				ProcessInstance instance = processInstanceMapper.selectByPrimaryKey(taskLog.getId());
				if (instance != null) {
					processDefinitionId = instance.getProcessDefinitionId();
				}
			}
			if (StringUtils.isNotBlank(processDefinitionId) && StringUtils.isNotBlank(contents)) {
				webSocketTaskLogServer.sendMessage(processDefinitionId, contents);
			}

		} catch (IOException e) {
			e.printStackTrace();
		}

	}

	public void queryProcessLogs(Session session, String message) {
		//{"type":"all/task","processId":"","taskId":""}
		//all--表示全部日志；task--表示节点日志；
		logger.info("push log msg: " + message);
		if (StringUtils.isBlank(message)) {
			return;
		}
		Map map = JSONUtils.toObjectMap(message);
		String type = MapUtils.getString(map, "type", "all");
		String processId = MapUtils.getString(map, "processId", "");
		String taskId = MapUtils.getString(map, "taskId", "");
		//日志内容
		String contents = "";
		//判断是不是通用流程的任务
		String flowType = processInstanceMapper.getFLowTypeByProcessId(processId);
		if ("commonFlow".equals(flowType)) {
			processCommonFlowLog(session, type, processId, taskId);
			//结束通用流程的日志发送
			return;

		} else {
			TaskLog taskLog = taskLogService.selectByPrimaryKey(processId);
			if (taskLog == null || StringUtils.isBlank(taskLog.getContents())) {
				return;
			}
			contents = taskLog.getContents();
		}
		if (StringUtils.isBlank(contents)) {
			return;
		}

		//如果是算子节点，获取算子名称
		String taskName = "";
		if ("task".equalsIgnoreCase(type)) {
			TaskInstance instance = taskInstanceMapper.selectByPrimaryKey(taskId);
			if (instance != null) {
				taskName = instance.getName();
			}
		}
		//设置每条日志推的最小间隔 (ms)
		int millis = ConfigUtils.getInstance().getInteger("dds.log.pushTime", 100);
		String[] lines = contents.split("\n");
		String lastTask = "";
		for (String line : lines) {
			if (!line.startsWith("\t")) {
				String[] logInfos = line.split(" ");
				if (logInfos.length >= 3) {
					lastTask = logInfos[2];
				}
			}
			try {
				line = line + "\n";
				if ("task".equalsIgnoreCase(type) && taskName.equalsIgnoreCase(lastTask)) {
					//算子节点日志
					webSocketTaskLogServer.sendMessage(session, line);
				} else if ("all".equalsIgnoreCase(type)) {
					//全部日志
					webSocketTaskLogServer.sendMessage(session, line);
					Thread.sleep(millis);
				}

			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

	private void processCommonFlowLog(Session session, String type, String processId, String taskId) {
		String contents = "";
		//获取所有日志
		if ("all".equals(type)) {
			if (StringUtils.isBlank(processId)) {
				return;
			}
			List<TaskInstance> list = taskInstanceMapper.getFlowTaskProcessByInstanceId(processId);
			for (TaskInstance instance : list) {
				TaskLog taskLog = taskLogService.selectByPrimaryKey(instance.getId());
				if (taskLog == null || StringUtils.isBlank(taskLog.getContents())) {
					continue;
				}
				contents = taskLog.getContents();
				sendSocketMessage(session, contents);
			}

		} else {
			if (StringUtils.isBlank(taskId)) {
				return;
			}
			TaskLog taskLog = taskLogService.selectByPrimaryKey(taskId);
			if (taskLog == null || StringUtils.isBlank(taskLog.getContents())) {
				return;
			}
			contents = taskLog.getContents();
			sendSocketMessage(session, contents);
		}
	}

	private void sendSocketMessage(Session session, String contents) {
		if (StringUtils.isBlank(contents)) {
			return;
		}
		String[] lines = contents.split("\n");
		for (String line : lines) {
			try {
				line = line + "\n";
				webSocketTaskLogServer.sendMessage(session, line);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public void queryProcessTaskStatus(Session session, String message) {
		//{"processId":"","processDefinitionId":""}
		//process--表示流程实例；
		logger.info("push process task msg: " + message);
		if (StringUtils.isBlank(message)) {
			return;
		}
		Map msgMap = JSONUtils.toObjectMap(message);
		String processId = MapUtils.getString(msgMap, "processId", "");
		String processDefinitionId = MapUtils.getString(msgMap, "processDefinitionId", "");

		ProcessInstance instance = processInstanceMapper.selectByPrimaryKey(processId);
		List<Map<String, Object>> list = new ArrayList<>();
		if (instance == null) {
			Map<String, Object> map = new HashMap<>();
			map.put("processId", processId);
			map.put("runningState", "1");
			map.put("id", "");
			map.put("state", "");
			String info = JSONUtils.toJsonString(map);
			sendTaskStatus(processDefinitionId, info);
		} else {
			Integer runningState = instance.getState();
			List<TaskInstance> taskList = taskInstanceMapper.getFlowTaskProcessByInstanceId(processId);
			if (!taskList.isEmpty()) {
				for (TaskInstance task : taskList) {
					Map<String, Object> map = convertToMap(task);
					map.put("processId", processId);
					//实例状态
					map.put("runningState", runningState);

					String info = JSONUtils.toJsonString(map);
					sendTaskStatus(processDefinitionId, info);
				}
			} else {
				Map<String, Object> map = new HashMap<>();
				map.put("processId", processId);
				map.put("runningState", runningState);
				map.put("id", "");
				map.put("state", "");
				String info = JSONUtils.toJsonString(map);
				sendTaskStatus(processDefinitionId, info);
			}
		}
	}

	private Map<String, Object> convertToMap(TaskInstance task) {
		Map<String, Object> map = new HashMap<>();
		if (task == null) {
			return map;
		}
		//算子状态
		Integer state = TaskStateUtils.convertTaskState(task.getState());
		map.put("state", state);
		String startTime = task.getStartTime() == null ? "" : DateUtils.getDateByString(task.getStartTime());
		map.put("startTime", startTime);
		String endTime = task.getEndTime() == null ? "" : DateUtils.getDateByString(task.getEndTime());
		map.put("endTime", endTime);
		map.put("id", task.getId());
		map.put("name", task.getName());
		map.put("stateName", TaskStateUtils.convertStateName(state));
		map.put("chartType", task.getChartType() == null ? 0 : task.getChartType());
		map.put("nodeType", task.getNodeType() == null ? 0 : task.getNodeType());
		return map;
	}

	private void sendTaskStatus(String processDefinitionId, String info) {
		try {
			webSocketTaskStatusServer.sendMessage(processDefinitionId, info);
			webSocketProcessStausServer.sendMessage(info);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void updateTaskStatus(TaskInstance taskInstance) {
		//运行实例更新节点状态
		//  0 -初始化； 1--开始运行；7--成功；6 --失败； 9--Killed
		taskInstanceMapper.updateByPrimaryKeySelective(taskInstance);
		//运行实例状态
		ProcessInstance instance = processInstanceMapper.selectByPrimaryKey(taskInstance.getProcessInstanceId());
		if (instance == null) {
			return;
		}
		Integer runningState = instance.getState();
		//totalTasks, finishedTasks
		Map<String, Object> map = taskInstanceMapper.getTaskStateCount(taskInstance.getProcessInstanceId());
		//处理达梦返回字段都是大写的问题
		if (map.containsKey("TOTALTASKS")) {
			Integer totalTasks = MapUtils.getInteger(map, "TOTALTASKS", 0);
			Integer finishedTasks = MapUtils.getInteger(map, "FINISHEDTASKS", 0);
			map.put("totalTasks", totalTasks);
			map.put("finishedTasks", finishedTasks);
		}
		TaskInstance task = taskInstanceMapper.selectByPrimaryKey(taskInstance.getId());
		Map<String, Object> taskMap = convertToMap(task);
		map.putAll(taskMap);
		map.put("processId", taskInstance.getProcessInstanceId());
		map.put("processDefinitionId", instance.getProcessDefinitionId());
		//实例状态
		map.put("runningState", runningState);

		String info = JSONUtils.toJsonString(map);
		sendTaskStatus(instance.getProcessDefinitionId(), info);
	}

	public void sendWebsocketTaskStatus(String processDefinitionId, String processId, Integer runningState) {

		//totalTasks, finishedTasks
		Map<String, Object> map = taskInstanceMapper.getTaskStateCount(processId);
		//处理达梦返回字段都是大写的问题
		if (map.containsKey("TOTALTASKS")) {
			Integer totalTasks = MapUtils.getInteger(map, "TOTALTASKS", 0);
			Integer finishedTasks = MapUtils.getInteger(map, "FINISHEDTASKS", 0);
			map.put("totalTasks", totalTasks);
			map.put("finishedTasks", finishedTasks);
		}
		map.put("processId", processId);
		map.put("processDefinitionId", processDefinitionId);
		map.put("id", "");
		map.put("state", "");
		map.put("runningState", runningState);
		map.put("runTime", DateUtils.dateToString(new Date()));
		String info = JSONUtils.toJsonString(map);
		sendTaskStatus(processDefinitionId, info);
	}

	public void updateProcessStatus(ProcessInstance processInstance) {
		if (1 == processInstance.getState()) {
			processInstance.setStartTime(new Date());
		}
		processInstanceMapper.updateByPrimaryKeySelective(processInstance);

		ProcessInstance instance = processInstanceMapper.selectByPrimaryKey(processInstance.getId());
		if (instance == null) {
			return;
		}
		//totalTasks, finishedTasks
		Map<String, Object> map = taskInstanceMapper.getTaskStateCount(processInstance.getId());
		//处理达梦返回字段都是大写的问题
		if (map.containsKey("TOTALTASKS")) {
			Integer totalTasks = MapUtils.getInteger(map, "TOTALTASKS", 0);
			Integer finishedTasks = MapUtils.getInteger(map, "FINISHEDTASKS", 0);
			map.put("totalTasks", totalTasks);
			map.put("finishedTasks", finishedTasks);
		}
		map.put("processId", processInstance.getId());
		map.put("processDefinitionId", instance.getProcessDefinitionId());
		map.put("id", "");
		map.put("state", "");
		map.put("runningState", processInstance.getState());
		map.put("runTime", DateUtils.dateToString(new Date()));
		String info = JSONUtils.toJsonString(map);
		sendTaskStatus(instance.getProcessDefinitionId(), info);
	}

	public void updateTaskAppId(TaskInstance taskInstance) {
		this.taskInstanceMapper.updateTaskAppIdByProcessInstanceId(taskInstance.getProcessInstanceId(), taskInstance.getAppLink());
		Map<String, Object> exParams = new HashMap<>(16);
		exParams.put("appid", taskInstance.getAppLink());
		ProcessInstance processInstance = new ProcessInstance();
		processInstance.setId(taskInstance.getProcessInstanceId());
		processInstance.setExtraParams(JSONUtils.toJson(exParams));
		processInstanceMapper.updateByPrimaryKeySelective(processInstance);
	}
}
